package com.atguigu.sql.query;

/*
* @date 2024/11/25 15:12
* 简单查询
* */

import com.atguigu.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink01_SimpleQuery {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //流表环境
        StreamTableEnvironment streamTableEnv = StreamTableEnvironment.create(env);

        //数据流
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {

                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Integer.valueOf(fields[1].trim()), Long.valueOf(fields[2].trim()));

                        }
                );

        //流转表
        //todo Schema 是用来定义表结构的一个对象。具体来说，它描述了表中的列及其数据类型，
        // 还包括基于表达式的虚拟列（如处理时间 PROCTIME() 和通过表达式转换的时间戳）以及水印策略。
        // 这个 Schema 对象随后被用于将一个 DataStream 转换成 Table API 中的 Table 对象。

        Schema schema = Schema.newBuilder()
                .column("id", "STRING")
                .column("vc", "INT")
                .column("ts", "BIGINT")
                .columnByExpression("pt", "PROCTIME()")
                .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                .watermark("et", "TO_TIMESTAMP_LTZ(ts,3)")
                .build();
        Table table = streamTableEnv.fromDataStream(ds, schema);
        streamTableEnv.createTemporaryView("t1",table); //将表对象注册成表

        //简单查询
        //1.select where
        //streamTableEnv.sqlQuery("select id, vc , ts from t1 where vc >= 100").execute().print();

        //2.with 字句，CTE
        //streamTableEnv.sqlQuery("with t2 as (select id, vc, ts from t1 where vc >=100) select vc from t2").execute().print();

        //3.去重,对key去重
        //streamTableEnv.sqlQuery("select distinct id from t1").execute().print();

        //4.order by 和 LIMIT
        //todo 当使用ts字段排序时报错 Sort on a non-time-attribute field is not supported.
        //order by: 只能使用时间属性字段 且 只能是升序
        //todo 这里只对时间进行排序，当插入时间小于输入时间，数据不显示
        //LIMIT: 只支持Batch模式
        //streamTableEnv.sqlQuery("select id,vc,ts from t1 order by et asc").execute().print();

        //5.sql Hits  是sql提示
       //String sourceTable =
       //        "create table t3(" +
       //                "id STRING," +
       //                "vc INT ," +
       //                "ts BIGINT" +
       //        " ) WITH ( " +
       //                "'connector'= 'kafka'," +
       //                "'topic' = 'topicA'," +
       //                "'properties.bootstrap.servers' = 'hadoop102:9092,hadoop103:9092,hadoop104:9092'," +
       //                "'properties.group.id' = 'flinksql240620'," +
       //                "'format' = 'csv'," +
       //                "'scan.startup.mode' = 'latest-offset'" +
       //        ")";
       //streamTableEnv.executeSql(sourceTable);

       // streamTableEnv.sqlQuery("select id ,vc ,ts from t3").execute().print();

        //6.集合操作
        streamTableEnv.executeSql("create view t4(s) as values ('c'),('a'),('b'),('c'),('d'),('a')");
        streamTableEnv.executeSql("create view t5(s) as values ('d'), ('e'), ('a'), ('b'), ('b')");

        //UNION 和UNION ALL ：UNION会去重
        //streamTableEnv.sqlQuery("select s from t4 UNION select s from t5").execute().print();
        //streamTableEnv.sqlQuery("select s from t5 UNION ALL select s from t5").execute().print();
        //todo 测试下executeSql() 与 sqlQuery().execute()效果是一样的
        //streamTableEnv.executeSql("select s from t4 UNION select s from t5").print();

        //INTERSECT 和 INTERSECT ALL（intersect：意为交叉的意思）
        //todo INTERSECT 是一个集合运算符，用于找出两个查询结果集的交集,并去重
        //streamTableEnv.sqlQuery("select s from t4 INTERSECT select s from t5").execute().print();
        //streamTableEnv.sqlQuery("select s from t4 INTERSECT ALL select s from t5").execute().print();

        //EXCEPT 和 EXCEPT ALL 差集
        //todo EXCEPT 是一个集合运算符，用于返回第一个查询结果集中存在但不在第二个查询结果集中的行。
        streamTableEnv.sqlQuery("select s from t4 EXCEPT select s from t5").execute().print();
        //streamTableEnv.sqlQuery("select s from t4 EXCEPT ALL select s from t5").execute().print();

        //7.分组聚合
        //streamTableEnv.sqlQuery("select id ,sum(vc) as sum_vc from t1 group by id ").execute().print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
